package com.nathan.mq.integration.service.impl;

import com.nathan.mq.integration.service.RedisService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RBucket;
import org.redisson.api.RLock;
import org.redisson.api.RMapCache;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.map.event.EntryExpiredListener;
import org.redisson.client.codec.Codec;
import org.redisson.codec.FstCodec;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @Description:
 * @Author: zhujunjie
 * @Date: 2021/12/9 17:20
 */
@Slf4j
@Component
public class RedisServiceImpl implements RedisService {

    @Value("${spring.profiles.active:dev}")
    private String env;
    private static final String PREFIX = "mq-integration:";
    private static final long LOCK_WAIT_TIME_MILL_SECOND = 3000L;
    private static final long LEASE_TIME_MILL_SECOND = 60 * 1000;

    @Resource
    private RedissonClient redissonClient;

    private String renameKey(String key) {
        return PREFIX + env + ":" + key;
    }

    @Override
    public <T> void addQueue(String name, T data) {
        addDelayedQueue(name, data, 1, TimeUnit.MILLISECONDS);
    }

    @Override
    public <T> void deleteDelayedQueue(String name) {
        final String key = renameKey(name);
        redissonClient.getDelayedQueue(redissonClient.getBlockingQueue(key)).delete();
    }

    @Override
    public <T> void addDelayedQueue(String name, T data, long delay, TimeUnit timeUnit) {
        final String key = renameKey(name);
        redissonClient.getDelayedQueue(redissonClient.getBlockingQueue(key)).offer(data, delay, timeUnit);
    }

    @Override
    public <T> RBlockingQueue<T> getBlockingQueue(String name) {
        final String key = renameKey(name);
        return redissonClient.getBlockingQueue(key);
    }

    @Override
    public <V> boolean set(String key, V val) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis set key:{}, forever", cacheKey);
            final RBucket<V> bucket = redissonClient.getBucket(cacheKey);
            bucket.set(val);
            return true;
        } catch (Exception e) {
            log.error("Redis exp key:{} forever {}", cacheKey, e);
            return false;
        }
    }

    @Override
    public <V> boolean set(String key, V val, long expired, TimeUnit timeUnit) {
        final String cacheKey = renameKey(key);
        try {
            log.info("Redis set key:{}, expired:{}, timeUnit:{}", cacheKey, expired, timeUnit);
            final RBucket<V> bucket = redissonClient.getBucket(cacheKey);
            bucket.set(val, expired, timeUnit);
            return true;
        } catch (Exception e) {
            log.error("Redis exp key:{} expired:{} {}", cacheKey, expired, e);
            return false;
        }
    }

    @Override
    public boolean delete(String key) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis delete key:{}", cacheKey);
            redissonClient.getBucket(cacheKey).delete();
            return true;
        } catch (Exception e) {
            log.error("Redis exp delete key:{} {}", cacheKey, e);
            return false;
        }
    }

    @Override
    public <R> R get(String key) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis get key:{}", cacheKey);
            final RBucket<R> bucket = redissonClient.getBucket(cacheKey);
            return bucket.get();
        } catch (Exception e) {
            log.error("Redis exp get key:{} {}", cacheKey, e);
            return null;
        }
    }

    @Override
    public <R> R get(String key, Function<String, R> function, long timeToLive, TimeUnit timeUnit) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis get key:{}", cacheKey);
            final RBucket<R> bucket = redissonClient.getBucket(cacheKey);
            final R r = bucket.get();
            if (r == null) {
                redissonClient.getBucket(cacheKey).set(function.apply(key), timeToLive, timeUnit);
            }
            return r;
        } catch (Exception e) {
            log.error("Redis exp get key:{} {}", cacheKey, e);
            return null;
        }
    }

    @Override
    public RAtomicLong getAtomicLong(String key) {
        final RAtomicLong atomicLong = redissonClient.getAtomicLong(renameKey(key));
        // 默认30天过期
        atomicLong.expire(30, TimeUnit.DAYS);
        return atomicLong;
    }

    @Override
    public RMapCache<String, Object> getMapCache(String name) {
        return redissonClient.getMapCache(name);
    }

    @Override
    public Integer addExpired(String name, Object value, Long intervalMils, EntryExpiredListener expiredListener) {
        final RMapCache<String, Object> cache = getMapCache(PREFIX + name);
        final long minInterval = 0 == intervalMils ? 1000 : intervalMils;
        cache.put(name, value, minInterval, TimeUnit.MILLISECONDS);
        return cache.addListener(expiredListener);
    }

    @Override
    public void removeExpired(String name, Integer listener) {
        final RMapCache<String, Object> cache = getMapCache(PREFIX + name);
        cache.removeListener(listener);
    }

    @Override
    public void publish(String topicName, Object value) {
        redissonClient.getTopic(renameKey(topicName), new FstCodec()).publish(value);
    }

    @Override
    public void publish(String topicName, Object value, Codec codec) {
        redissonClient.getTopic(renameKey(topicName), codec).publish(value);
    }

    @Override
    public RTopic getTopic(String topicName, Codec codec) {
        return redissonClient.getTopic(renameKey(topicName), codec);
    }

    /**
     * 根据时间按天生成自增序号
     */
    @Override
    public String getIncrementNumDays(String key, int number) {
        // 不存在准备创建 键值对
        final RAtomicLong rAtomicLong = redissonClient.getAtomicLong(renameKey(key));
        final Long counter = rAtomicLong.incrementAndGet();
        // 初始设置过期时间
        if (counter == 1) {
            log.info("自增key:{},设置过期时间为1天", key);
            // 单位天
            rAtomicLong.expire(1, TimeUnit.DAYS);
        }
        return StringUtils.leftPad(counter.toString(), number, "0");
    }

    /**
     * 防重机制 方法加锁
     *
     * @param key      不为空
     * @param supplier 不为空
     */
    @Override
    public <T> T lock(String key, Supplier<T> supplier) {
        final RLock lock = redissonClient.getLock(renameKey(key));
        if (lock.tryLock()) {
            try {
                return supplier.get();
            } finally {
                lock.unlock();
            }
        } else {
            throw new RuntimeException("请勿重复操作！");
        }
    }

}
